package com.shujia;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;

public class Demo4Filter {
    Connection conn;
    Admin admin; // 对表结构的修改或者增加删除表
    Table students; // 对表中的数据进行操作
    TableName studentsT = TableName.valueOf("students");

    public void printScannerWithFilter(Filter filter) throws IOException {
        Scan scan = new Scan();

        scan.setFilter(filter);
        ResultScanner scanner = students.getScanner(scan);
        for (Result result : scanner) {
            String id = Bytes.toString(result.getRow());
            String resultStr = id + "\t";
            for (Cell cell : result.listCells()) {
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                resultStr = resultStr + value + "\t";
            }

            System.out.println(resultStr);
        }
    }

    @Before
    public void init() throws IOException {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "master,node1,node2");
        conn = ConnectionFactory.createConnection(conf);
        admin = conn.getAdmin();
        students = conn.getTable(studentsT);

    }

    // RegexStringComparator + ValueFilter
    // ValueFilter 会将比较器作用到所有的列值上，也就相当于作用在每一个cell上
    // 把符合比较器规则的cell过滤出来，不符合的直接丢弃
    @Test
    public void RegexStringComparatorValueFilter() throws IOException {


        RegexStringComparator regexStringComparator = new RegexStringComparator(".*班.*");
        ValueFilter valueFilter = new ValueFilter(CompareFilter.CompareOp.EQUAL, regexStringComparator);

        printScannerWithFilter(valueFilter);
    }

    @Test
    // 单列值过滤器SingleColumnValueFilter
    // 会返回满足条件的cell所在行的所有cell的值（即会返回一行数据）
    public void SingleColumnValue() throws IOException {

        SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(
                "info".getBytes()
                , "clazz".getBytes()
                , CompareFilter.CompareOp.EQUAL
                , new RegexStringComparator("^文科.*班$"));

        printScannerWithFilter(singleColumnValueFilter);


    }

    @Test
    // 单列值排除过滤器
    // 作用同单列值过滤器类似，但是最后返回的结果会排除指定的比较列的数据
    public void SingleColumnValueExclude() throws IOException {

        SingleColumnValueExcludeFilter singleColumnValueExcludeFilter = new SingleColumnValueExcludeFilter(
                "info".getBytes()
                , "clazz".getBytes()
                , CompareFilter.CompareOp.EQUAL
                , new RegexStringComparator("^文科.*班$"));

        printScannerWithFilter(singleColumnValueExcludeFilter);


    }

    @Test
    // 把包含 a 的 所有的列 的 所有的数据 过滤出来，即输出（age、name、clazz三列的数据）
    // 因为gender这个列名不包含 a 所以该列数据不会被输出
    public void SubstringComparatorQualifierFilter() throws IOException {
        SubstringComparator comparator = new SubstringComparator("a");
        QualifierFilter qualifierFilter = new QualifierFilter(CompareFilter.CompareOp.EQUAL, comparator);

        printScannerWithFilter(qualifierFilter);


    }

    @Test
    // 过滤出id以 150010010 开头的所有的学生信息
    // RowFilter会返回符合条件的Rowkey所在的整行数据
    public void BinaryPrefixComparatorRowFilter() throws IOException {
        BinaryPrefixComparator binaryPrefixComparator = new BinaryPrefixComparator("150010010".getBytes());
        RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL, binaryPrefixComparator);

        printScannerWithFilter(rowFilter);


    }

    @Test
    // 过滤出id以 150010010 开头的所有的学生信息
    // PrefixFilter rowkey前缀过滤器 相当于 RowFilter + BinaryPrefixFilter
    public void RowKeyPrefixFilter() throws IOException {
        PrefixFilter prefixFilter = new PrefixFilter("150010010".getBytes());

        printScannerWithFilter(prefixFilter);


    }

    @Test
    // 分页过滤器PageFilter
    // 实际上相当于scan.setLimit()
    // 效率非常低，因为每次都需要遍历当前页面之前的所有数据
    public void PageFilter() throws IOException {
        int pageSize = 10;
        int page = 4;

        // 计算出当前页面之前的记录数 + 1
        int current_page_first_rk = (page - 1) * pageSize + 1;

        Scan scan = new Scan();

        PageFilter pageFilter = new PageFilter(current_page_first_rk);

        scan.setFilter(pageFilter);
        byte[] start_rk = null;

        // 取当前页面的第一条数据的Rowkey
        for (Result result : students.getScanner(scan)) {
            start_rk = result.getRow();
        }

        Scan scan1 = new Scan();
        scan1.withStartRow(start_rk);
//        PageFilter pageFilter1 = new PageFilter(pageSize);
//        scan1.setFilter(pageFilter1);
        scan1.setLimit(pageSize); // 可以替代PageFilter

        for (Result result : students.getScanner(scan1)) {
            String id = Bytes.toString(result.getRow());
            String resultStr = id + "\t";
            for (Cell cell : result.listCells()) {
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                resultStr = resultStr + value + "\t";
            }

            System.out.println(resultStr);
        }


    }

    @Test
    // 如果要做分页 最好的方式是 合理的设计Rowkey
    public void PageFilterTest2() throws IOException {
        int baseId = 1500100000;
        int pageSize = 10;
        int page = 4;

        // 如果rowkey设计的合理 可以直接计算出每一页的start_rowkey 和 end_rowkey
        int current_page_first_rk = baseId + (page - 1) * pageSize + 1;
        int current_page_last_rk = current_page_first_rk + pageSize;

        Scan scan = new Scan();
        scan.withStartRow(String.valueOf(current_page_first_rk).getBytes());
        scan.withStopRow(String.valueOf(current_page_last_rk).getBytes());

        for (Result result : students.getScanner(scan)) {
            String id = Bytes.toString(result.getRow());
            String resultStr = id + "\t";
            for (Cell cell : result.listCells()) {
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                resultStr = resultStr + value + "\t";
            }

            System.out.println(resultStr);
        }

    }

    @Test
    // 过滤出 age >= 23 并且 gender = '男' 并且 rowkey以 150010010开头 的所有的学生信息
    public void combineFilter() throws IOException {
        // age >= 23
        SingleColumnValueFilter filter1 = new SingleColumnValueFilter(
                "info".getBytes(),
                "age".getBytes(),
                CompareFilter.CompareOp.GREATER_OR_EQUAL,
                "23".getBytes());

        // gender = '男'
        SingleColumnValueFilter filter2 = new SingleColumnValueFilter(
                "info".getBytes(),
                "gender".getBytes(),
                CompareFilter.CompareOp.EQUAL,
                "男".getBytes());

        // rowkey以 150010010 开头
        PrefixFilter filter3 = new PrefixFilter("150010010".getBytes());

        FilterList filterList = new FilterList();

        filterList.addFilter(filter1);
        filterList.addFilter(filter2);
        filterList.addFilter(filter3);

        printScannerWithFilter(filterList);

    }

    @After
    public void closed() throws IOException {
        conn.close();
    }
}
